package com.etl.dataflow.executor.communication;

import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.etl.dataflow.common.entity.TaskInfo;
import com.etl.dataflow.common.response.Code;
import com.etl.dataflow.common.util.ExceptionUtil;
import com.etl.dataflow.common.util.ResponseUtil;
import com.etl.dataflow.executor.core.TaskRunner;
import com.etl.dataflow.executor.entity.ExecutorTaskInfo;
import com.etl.dataflow.executor.util.SchedulerInfoUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Queue;
import java.util.concurrent.*;

/**
 * 任务超时检测，对于超时的任务直接结束
 *
 * @author dx
 * @since 2022/12/4
 **/
@Component
@Slf4j
public class TaskTimeoutChecker {

    private static final ExecutorService LOOP_EXECUTOR_SERVICE = new ThreadPoolExecutor(
            1,
            1,
            10,
            TimeUnit.SECONDS,
            new LinkedBlockingDeque<>(),
            r -> new Thread(r, "TaskTimeoutCheckerThread"),
            new ThreadPoolExecutor.DiscardOldestPolicy()
    );

    @Resource
    private SchedulerInfoUtil schedulerInfoUtil;

    @Value("${executor.task.TIMEOUT_MINUTES}")
    private Integer timeOutMinutes;

    @PostConstruct
    private void loop() {
        LOOP_EXECUTOR_SERVICE.submit(() -> {
            //noinspection InfiniteLoopStatement
            while (true) {
                try {
                    checkTimeoutTask();
                    TimeUnit.SECONDS.sleep(1L);
                } catch (Throwable e) {
                    e.printStackTrace();
                    log.error(ExceptionUtil.getFullStackTrace(e));
                }
            }
        });
    }

    private void checkTimeoutTask() {
        for (TaskRunner taskRunner : TaskRunnerHolder.getInstance().getAllTaskRunner()) {
            long taskLaunchTime = taskRunner.getTaskLaunchTime();
            if (taskLaunchTime == 0) {
                continue;
            }
            long timeGap = millisecond2minute(System.currentTimeMillis() - taskLaunchTime);
            if (timeGap > timeOutMinutes) {
                taskRunner.shutdown();
                ExecutorTaskInfo executorTaskInfo = taskRunner.getExecutorTaskInfo();
                log.debug("任务超时：taskRecordId: {}, dstTable: {}",
                        executorTaskInfo.getTaskRecordId(), executorTaskInfo.getTaskParam().getDstTableName());
            }
        }
        // 删除异常任务
        TaskRunnerHolder.getInstance().removeIllegalTasks();
        // 拿到删除的任务信息，通知调度中心任务超时
        Queue<ExecutorTaskInfo> queue = TaskRunnerHolder.getInstance().getTimeoutTaskQueue();
        while (!queue.isEmpty()) {
            ExecutorTaskInfo taskInfo = queue.poll();
            if (taskInfo != null) {
                notifySchedulerTaskTimeout(taskInfo);
            }
        }
    }

    private long millisecond2minute(long millisecond) {
        return millisecond / 1000 / 60;
    }


    private void notifySchedulerTaskTimeout(ExecutorTaskInfo taskInfo) {
        boolean success = tryNotifySchedulerTaskTimeout(taskInfo);
        if (success) {
            return;
        }
        // 如果失败了再尝试三次
        for (int i = 0; i < com.etl.dataflow.executor.communication.service.ExecutorService.MAX_RETRY_NUM; i++) {
            if (tryNotifySchedulerTaskTimeout(taskInfo)) {
                return;
            }
        }
    }

    private boolean tryNotifySchedulerTaskTimeout(TaskInfo taskInfo) {
        String subUrl = String.format("/state/timeout/%s/%s/%s", ExecutorMark.getInstance().getUuid(),
                taskInfo.getId(), taskInfo.getTaskRecordId());
        try (HttpResponse httpResponse = HttpUtil
                .createPost(schedulerInfoUtil.getSchedulerUrl() + subUrl)
                .execute()) {
            String body = httpResponse.body();
            JSONObject jsonObject = JSONUtil.parseObj(body);
            boolean success = false;
            if (ResponseUtil.isSuccess(jsonObject) && jsonObject.getJSONObject("data") != null) {
                JSONObject data = jsonObject.getJSONObject("data");
                Boolean result = data.getBool("result");
                if (result != null && result) {
                    log.debug("taskRecordId: " + taskInfo.getTaskRecordId() + "，通知调度中心任务状态为超时，返回结果：" + body);
                    success = true;
                }
            } else if (ResponseUtil.codeEquals(jsonObject, Code.TASK_STATE_UNMODIFIABLE)) {
                // 调度器已告知任务状态不可更改，重试也是在做无用功，不如返回成功避免多次重试
                success = true;
            }
            if (!success) {
                log.debug("taskRecordId: " + taskInfo.getTaskRecordId() + "，通知调度中心任务状态为超时，返回结果：" + body);
            }
            return success;
        }
    }
}
